package com.yangcangclub.mq.domain;

import com.yangcangclub.mq.base.MQBaseDomain;
import com.yangcangclub.mq.bean.MQOffsetInfo;
import com.google.common.collect.Maps;
import com.yangcangclub.commons.exception.ApolloException;
import com.yangcangclub.commons.result.MessageEnum;
import org.springframework.stereotype.Component;

import java.util.Map;


@Component
public class MQOffsetProfile extends MQBaseDomain {

    /**
     * 获取offset
     * @param consumer
     * @param partition
     * @param topic
     * @param tag
     * @return
     */
    public MQOffsetInfo getMQOffset(String consumer,int partition,String topic, String tag){

        Map<String,Object> map = Maps.newHashMap();
        map.put("consumer",consumer);
        map.put("partition_",partition);
        map.put("topic",topic);
        map.put("tag",tag);

        MQOffsetInfo mqOffsetInfo = super.getSlaveJdbc().selectOne("mq.MQMapper.selectConsumerOffset",map);
        if(null==mqOffsetInfo) {
            MQOffsetInfo mqOffsetInfo1 = new MQOffsetInfo();
            mqOffsetInfo1.setConsumer(consumer);
            mqOffsetInfo1.setPartition_(partition);
            mqOffsetInfo1.setTopic(topic);
            mqOffsetInfo1.setTag(tag);
            mqOffsetInfo1.setOffset(0);
            super.getMasterJdbc().insert("mq.MQMapper.insertConsumerOffset",mqOffsetInfo1);
            return mqOffsetInfo1;
        } else {
            return mqOffsetInfo;
        }
    }

    /**
     * 更新offset
     * @param id
     * @param offset
     * @return
     */
    public boolean updateMqOffset(int id,int offset) {

        Map<String,Object> map = Maps.newHashMap();
        map.put("id", id);
        map.put("offset", offset);

        int i =  super.getMasterJdbc().update("mq.MQMapper.updateConsumerOffset", map);

        if(i != 1) {
            //回滚
            throw new ApolloException(MessageEnum.DATABASE_ERROR);
        }

        return true;
    }

}
